package streamAPI.sink;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;

/**
 * @author yue.cao
 * @since 10-28-2020
 */
@Slf4j
public class MyRedisHashSink extends RichSinkFunction<Tuple3<String, String, String>> {

	private transient JedisCluster jedis;

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		jedis = new JedisCluster(new HostAndPort("10.199.186.255", 6381));
		ParameterTool tool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
		log.warn("config detail:[{}]", tool.toMap().toString());
	}

	@Override
	public void invoke(Tuple3<String, String, String> value, Context context) throws Exception {
		jedis.hset(value.f0, value.f1, value.f2);
	}

	@Override
	public void close() throws Exception {
		super.close();
		jedis.close();
	}
}
